package com.idanchuang.support.etcache.pubsub.amqp;

import com.idanchuang.support.etcache.pubsub.AbstractRemoteEventSubscriber;
import com.idanchuang.support.etcache.pubsub.SubEventQueue;
import com.idanchuang.support.etcache.remote.codec.RemoteKeyCodec;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * 基于AMQP实现的缓存变更事件监听器
 * @author yjy
 * Created at 2021/12/22 12:44 下午
 */
public class AmqpRemoteEventSubscriber extends AbstractRemoteEventSubscriber implements Consumer {

    private static final Logger logger = LoggerFactory.getLogger(AmqpRemoteEventSubscriber.class);

    private final ConnectionFactory connectionFactory;
    private final String exchange;
    private String queue;
    private Connection connection;
    private Channel channel;
    private volatile boolean shutdown = false;

    public AmqpRemoteEventSubscriber(RemoteKeyCodec remoteKeyCodec,
                                     SubEventQueue subEventQueue,
                                     String appId,
                                     ConnectionFactory connectionFactory) {
        super(remoteKeyCodec, subEventQueue);
        try {
            this.connectionFactory = connectionFactory;
            this.exchange = "etcache-event-" + appId.toLowerCase();
            this.connection = this.connectionFactory.newConnection();
            this.init();
        } catch (Exception e) {
            throw new RuntimeException("AmqpRemoteEventSubscriber init failed", e);
        }
    }

    private void init() throws Exception {
        this.channel = connection.createChannel();
        this.channel.exchangeDeclare(this.exchange, "fanout");
        this.queue = this.channel.queueDeclare().getQueue();
        this.channel.queueBind(queue, this.exchange, "");
        this.channel.basicConsume(queue, true, this);
        logger.info("AmqpRemoteEventSubscriber init finish > exchange: {}, queue: {}",
                exchange, queue);
    }

    @Override
    public void destroy() {
        try {
            if (!this.shutdown) {
                this.shutdown = true;
                if (this.connection != null && this.connection.isOpen()) {
                    this.connection.close();
                }
                logger.info("AmqpRemoteEventSubscriber destroy finish > exchange: {}, queue: {}",
                        exchange, queue);
            }
        } catch (Exception e) {
            logger.warn("connection close failed, cause: {}", e.getMessage());
        }
    }

    /**
     * 重连amqp
     */
    private void reconnect() {
        while (!this.channel.isOpen() && !shutdown) {
            try {
                Thread.sleep(1000L);
                if (!this.connection.isOpen()) {
                    this.connection = this.connectionFactory.newConnection();
                }
                this.init();
            } catch (InterruptedException e) {
                logger.warn("reconnect amqp cancel, cause thread is interrupted", e);
                break;
            } catch (Exception e) {
                logger.error("reconnect amqp failed, will retry after 1s..", e);
            }
        }
    }

    @Override
    public void handleConsumeOk(String consumerTag) {

    }

    @Override
    public void handleCancelOk(String consumerTag) {

    }

    @Override
    public void handleCancel(String consumerTag) throws IOException {
        reconnect();
    }

    @Override
    public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
        reconnect();
    }

    @Override
    public void handleRecoverOk(String consumerTag) {

    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        handle(body);
    }
}
